package club.westudy.distributed.lock.internal.etcd;

import club.westudy.distributed.lock.DistributedLockService;
import club.westudy.distributed.lock.internal.InternalLockExecutorService;
import io.etcd.jetcd.ByteSequence;
import io.etcd.jetcd.Client;
import io.etcd.jetcd.KV;
import io.etcd.jetcd.api.KeyValue;
import io.etcd.jetcd.api.PutResponse;
import io.etcd.jetcd.options.GetOption;
import io.etcd.jetcd.options.PutOption;

import java.nio.charset.Charset;
import java.util.List;
import java.util.concurrent.TimeUnit;

public class EtcdInternalLockExecutorService implements InternalLockExecutorService {
    Client client;
    KV kvClient;

    @Override
    public void initial(long guardScanTimeMs) {
        client = Client.builder().endpoints("http://localhost:2379").build();
        kvClient = client.getKVClient();


    }

    @Override
    public void destroy() {

    }

    @Override
    public DistributedLockService.CriticalRegionContext acquire(String feature, long maxLeaseTimeMs, long tryLockWaitTimeMs) {
        try {
            // 对String类型的key-value进行类型转换
            ByteSequence key = ByteSequence.from("test_key", Charset.forName("UTF-8"));
            ByteSequence value = ByteSequence.from("test_value", Charset.forName("UTF-8"));
            PutResponse putResponse = kvClient
                    .put(key,
                            value,
                            PutOption.newBuilder().withLeaseId(leaseId).build())
                    .get(10, TimeUnit.SECONDS);

            // 获取自己加锁操作的Revision号
            revisionOfMyself = putResponse.getHeader().getRevision();
        } catch (InterruptedException e1) {
            System.out.println("[error]: lock operation failed:" + e1);
        }

        try {
            // lockName作为前缀，取出所有键值对，并且根据Revision进行升序排列，版本号小的在前
            List<KeyValue> kvList = kvClient.get(ByteSequence.from(lockName),
                    GetOption.newBuilder().withPrefix(ByteSequence.from(lockName))
                            .withSortField(GetOption.SortTarget.MOD).build())
                    .get().getKvs();

            // 如果自己的版本号最小，则表明自己持有锁成功，否则进入监听流程，等待锁释放
            if (revisionOfMyself == kvList.get(0).getModRevision()) {
                System.out.println("[lock]: lock successfully. [revision]:" + revisionOfMyself);
                // 加锁成功，返回实际存储于Etcd中的key
                return strBufOfRealKey.toString();
            } else {
                // 记录自己加锁操作的前一个加锁操作的索引，因为只有前一个加锁操作完成并释放，自己才能获得锁
                int preIndex = 0;
                for (int index = 0; index < kvList.size(); index++) {
                    if (kvList.get(index).getModRevision() == revisionOfMyself) {
                        preIndex = index - 1;// 前一个加锁操作，故比自己的索引小1
                    }
                }
                // 根据索引，获得前一个加锁操作对应的key
                ByteSequence preKeyBS = kvList.get(preIndex).getKey();

                // 创建一个Watcher，用于监听前一个key
                Watcher watcher = client.getWatchClient().watch(preKeyBS);
                WatchResponse res = null;
                // 监听前一个key，将处于阻塞状态，直到前一个key发生delete事件
                // 需要注意的是，一个key对应的事件不只有delete，不过，对于分布式锁来说，除了加锁就是释放锁
                // 因此，这里只要监听到事件，必然是delete事件或者key因租约过期而失效删除，结果都是锁被释放
                try {
                    System.out.println("[lock]: keep waiting until the lock is released.");
                    res = watcher.listen();
                } catch (InterruptedException e) {
                    System.out.println("[error]: failed to listen key.");
                }

                // 为了便于读者理解，此处写一点冗余代码，判断监听事件是否为DELETE，即释放锁
                List<WatchEvent> eventlist = res.getEvents();
                for (WatchEvent event : eventlist) {
                    // 如果监听到DELETE事件，说明前一个加锁操作完成并已经释放，自己获得锁，返回
                    if (event.getEventType().toString().equals("DELETE")) {
                        System.out.println("[lock]: lock successfully. [revision]:" + revisionOfMyself);
                        return strBufOfRealKey.toString();
                    }
                }
            }
        } catch (InterruptedException | ExecutionException e) {
            System.out.println("[error]: lock operation failed:" + e);
        }

        return strBufOfRealKey.toString();
    }

    @Override
    public void release(DistributedLockService.CriticalRegionContext criticalRegionContext) {

    }
}
